package org.example.flinksql.test;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

// total 60000 records
// start time: 10:53:24,964
// end time: 10:53:54,779

public class JoinAndCopySingleTable {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

        // CREATE TEMPORARY TABLE t_enterprise
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_enterprise (" +
                "eid STRING," +
                "name STRING," +
                "credit_no STRING," +
                "category_new INT," +
                "regist_capi INT," +
                "actual_capi INT," +
                "status INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_enterprise'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // CREATE TEMPORARY TABLE t_address
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_address (" +
                "eid STRING," +
                "address STRING," +
                "postcode STRING," +
                "u_tags INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_address'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");
        // CREATE TEMPORARY TABLE t_address
        // TODO:Sink必须设置主键，当主键冲突时会自动更新数据，如果不设置程序会报错
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_enterprise_info (" +
                "eid STRING," +
                "name STRING," +
                "credit_no STRING," +
                "category_new INT," +
                "regist_capi INT," +
                "actual_capi INT," +
                "status INT," +
                "address STRING," +
                "postcode STRING," +
                "PRIMARY KEY (eid) NOT ENFORCED) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_enterprise_info'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // 一、直接内关联
//        Table query = tbEnv.sqlQuery("select a.*,b.address,b.postcode from t_enterprise as a INNER JOIN t_address as b ON a.eid = b.eid");
//        query.executeInsert("t_enterprise_info");


        // 二、查询eid进行关联
        // 0.获取enterprise列表
        // 1.从enterprise列表获取eid列表
        // 2.获取eid列表对应的address列表
        // 3.enterprise列表join获取的address列表
        // 4.写入enterprise_info表
        Table t_enterprise = tbEnv.from("t_enterprise").select($("*"));

        Table eids = t_enterprise.select($("eid"));

        Table t_address = tbEnv.from("t_address")
                .where($("eid").in(eids))
                .select($("eid"), $("address"), $("postcode"))
                .as("eid2,address,postcode");

        Table joinTable = t_enterprise.join(t_address).where($("eid").isEqual($("eid2")))
                .select($("eid"), $("name"), $("credit_no"), $("category_new"),
                        $("regist_capi"), $("actual_capi"), $("status"),
                        $("address"), $("postcode"))
                .as("eid,name,credit_no,category_new,regist_capi,actual_capi,status,address,postcode");

        joinTable.executeInsert("t_enterprise_info");


        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
